package cn.jly.flink.connector;

import com.alibaba.fastjson.JSON;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

import java.util.UUID;
import java.util.concurrent.atomic.AtomicLong;

/**
 * redis sink
 *
 * @author lanyangji
 * @create 2020-09-03 20:35
 */
public class RedisConnectorApp {
    /**
     * id生成器
     */
    private static final AtomicLong ID_GENERATOR = new AtomicLong(1000);

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<JdbcConnectorApp.Person> dataStreamSource = env.fromElements(
                new JdbcConnectorApp.Person(ID_GENERATOR.incrementAndGet(), UUID.randomUUID().toString().substring(0, 5)),
                new JdbcConnectorApp.Person(ID_GENERATOR.incrementAndGet(), UUID.randomUUID().toString().substring(0, 5)),
                new JdbcConnectorApp.Person(ID_GENERATOR.incrementAndGet(), UUID.randomUUID().toString().substring(0, 5)),
                new JdbcConnectorApp.Person(ID_GENERATOR.incrementAndGet(), UUID.randomUUID().toString().substring(0, 5))
        );

        // flink config
        FlinkJedisPoolConfig flinkJedisPoolConfig = new FlinkJedisPoolConfig.Builder().setHost("hadoop101").setPort(6379).build();

        // sink to redis
        dataStreamSource.addSink(new RedisSink<>(flinkJedisPoolConfig, new MyRedisMapper()));

        // redis-cli > config set protected-mode "no"
        // redis-cli > config set stop-writes-on-bgsave-error no
        env.execute("RedisConnectorApp");
    }

    public static class MyRedisMapper implements RedisMapper<JdbcConnectorApp.Person> {

        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.HSET, "Person_hset");
        }

        @Override
        public String getKeyFromData(JdbcConnectorApp.Person person) {
            return person.getId().toString();
        }

        @Override
        public String getValueFromData(JdbcConnectorApp.Person person) {
            return JSON.toJSONString(person);
        }
    }
}
